package com.example.dayu521.sg_oss.controller;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.example.dayu521.sg_oss.conf.AmqpCnf;
import com.example.dayu521.sg_oss.conf.RedisC;
import com.example.dayu521.sg_oss.domain.NumSplit;
import com.example.dayu521.sg_oss.service.NumSplitService;
import com.example.dayu521.sg_oss.service.OrderNumSource;
import com.example.dayu521.sg_oss.util.RedisHashBe;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Component
@RefreshScope
public class BaseOrderRecover {

    @Autowired
    private NumSplitService numSplitService;

    @Autowired
    private OrderNumSource orderNumSource;

    @Value("${sg.num.up_bound}")
    private Long up_bound;

    private final AtomicLong at=new AtomicLong();

    //todo 限流
    //todo 如何检测到数据不一致呢

    /** todo
     * 因为我们在冲突时候继续进行后面的操作,所以竞争冲突的概率增加,而且也完全不能判断数据不一致
     * 可以在冲突的地方直接退出.
     * 这样,我们期待那些在其他节点并发的冲突操作将会正确完成工作.否则,客户端很可能继续重试
     * 多次重试后,如果连续地出现冲突后退出,那么可以检查,是否出现了数据不一致,导致了流程无法继续
     */
    @RabbitListener(queues = {AmqpCnf.order_queue_incrby})
    public void incrby(NumSplit old) {
        //这里是客户端根据某个快照状态发送的消息
        //所以这里的恢复是快照状态无关的

        log.info("at: {}",at.incrementAndGet());

        var node_index=Long.parseLong(old.getNode_id());

        //假定消息是如实发出的,不是构造的.

        //根本是通过互斥状态来保证恢复是正确的
        //可以通过检查一些状态,尝试减少一部分竞争

        //获取当前快照状态
        var rxxx = (NumSplit)orderNumSource.pull_from(node_index, "sg_oss.incrby");
        //快照状态不存在.redis出现数据丢失.
        //状态不一致,也可以提前终止恢复
        if (rxxx == null) {
            log.error("""
                    一致性状态改变,当前恢复终止: redis中不存在NumSplit数据
                        ==> 重新获取NumSplit时,之前的NumSplit一定存在""");
            return;
        }
        //1.对比快照状态与消息描述.如果描述在快照状态之前,那就没必要恢复
        if(rxxx.getBase()>old.getBase()){
            log.info("过期恢复");
            return;
        }

        //下面是真正会并发的时候,遇到并发行为时,不会提前退出,而是继续进行之后的步骤
        //数据库记录,结束rxxx
        numSplitService.update(new UpdateWrapper<NumSplit>().set("status",rxxx.getId()).eq("id",rxxx.getId()));
        var uuid = UUID.randomUUID().toString();

        //初始化新的NumSplit
        NumSplit obj = new NumSplit(
                null, "xxx1:" + uuid, up_bound, 0l, rxxx.getNode_id(), 0l, new Date()
        );
        //数据库假定状态不会丢失
        var max_one = numSplitService.getOne(new QueryWrapper<NumSplit>()
                .apply("base=(select max(base) from t_num_split)"), true);
//        if (max_one == null) {
//            log.error("一致性状态改变,当前恢复终止: 数据库数据不存在任何数据,致命错误. ==> 前提,数据库中一定可以查到当前最大的记录base");
//            return ;
//        } else {
            obj.setBase(max_one.getBase() + max_one.getUp_bound());
//        }
        //注意,检测到并发执行时产生冲突,仍然进行之后的动作

        //尝试记录到数据库
        boolean save_ok = false;
        try {
            save_ok=numSplitService.save(obj);
        }catch (DuplicateKeyException e){

            //重复的原因可能是node_id与status列 唯一约束 => 并发恢复同一个redis节点
            //或者 base列的唯一约束  => 当前base由其他节点竞争成功

            //失败也继续进行
            save_ok=true;
        }
        if (save_ok) {

            var xxx=numSplitService.getOne(new QueryWrapper<NumSplit>()
                            .eq("status", 0)
                            .eq("base",obj.getBase())
                            .eq("node_id",obj.getNode_id()));
            //base由当前节点竞争到
            if(xxx!=null) {
                orderNumSource.save_not_exist(node_index, xxx.getIncr_key(), xxx.getUp_bound());

                //如果老的xxx.getIncr_key()对应的值没有丢失,客户端获取的数字是不合法的(负数),会触发下一次的恢复.
                //如果之前xxx.getIncr_key()对应的值也丢失了,则redis的递减操作会默认从0递减,那么递减后仍然是不合法的值.这样也触发了下一次恢复

                //这里提供了一个新的判断操作顺序的思路,客户端判断,而非服务本身判断.

                //并发问题是:并发的修改redis时,存在老的修改会覆盖新的修改的情况.
                //使用lua script阻止并发修改问题的时序问题
                orderNumSource.save_to(node_index, "sg_oss.incrby", xxx,xxx.getBase());
            }else{
                var onex = numSplitService.getOne(new QueryWrapper<NumSplit>()
                        .eq("status", 0)
                        .eq("node_id",obj.getNode_id()));
                if(onex!=null){//node_id和status上冲突了
                    orderNumSource.save_not_exist(node_index, onex.getIncr_key(), onex.getUp_bound());
                    orderNumSource.save_to(node_index, "sg_oss.incrby", onex,onex.getBase());
                    log.info("仅仅尝试恢复完成.此恢复目的是针对并发恢复过程中,之前恢复突然宕机的情况.");
                }else{//在base上冲突了
                    log.info("当前节点的恢复未竞争到base,等待下次恢复");
                }
            }
        } else {
            log.error("保存失败,当前恢复退出,等待下一次恢复");
        }
    }

    //首先需要判断触发恢复时的状态与当前状态是否对应
    //todo 限流
    @RabbitListener(queues = {AmqpCnf.order_queue_incrby_null})
    public void incrby_null(@Payload Long node_index) {

        //这里可能是初始状态的消息,但也可能是状态丢失的消息,所以恢复不一样
        //这里只恢复初始的空状态快照.丢失快照状态,会指出它,需要人工进行干预
        //这里的状态不但用来减少并发,也用于终止恢复过程

        //检查恢复条件
        log.info("at: {}",at.incrementAndGet());
        //当前快照状态,如果存在,则当前是过期恢复
        var rxxx = orderNumSource.pull_from(node_index, "sg_oss.incrby");
        if (rxxx != null) {
            log.info("过期恢复,退出");
            return;
        }

        //1.处理并发条件
        // 2 采用悲观策略.如果采用乐观策略,检测到冲突,直接退出,那么正在处理的竞争成功的节点还可能失败,如果它真失败了,就造成永久失败,必须人工干预,恢复到初始错误状态

        //最原始情况,所有节点都没有运行过
        var max_val = numSplitService.getOne(new QueryWrapper<NumSplit>()
                .apply("base=(select max(base) from t_num_split)"), true);
        if (max_val == null) {
            var save_ok = false;
            NumSplit numSplit = new NumSplit(null, "xxx1:"+UUID.randomUUID().toString(), up_bound, 0l, Long.toString(node_index), 0l, new Date());
            //下面失败直接返回,以后就再也不会进到这个分支了
            //所有的并发行为都会在这里竞争出成功者
            try {
                save_ok = numSplitService.save(numSplit);
            } catch (DuplicateKeyException e) {
                log.info("原始插入失败,检测并发插入,跳过插入继续");
            }
            if (!save_ok) {
                log.error("数据库插入范围失败,稍后再试");
                //这里不是并发错误,是真正的错误,需要退出
                return;
            }
            var multi_keys= Map.of(numSplit.getIncr_key(), numSplit.getUp_bound(),"sg_oss.incrby", numSplit);
            orderNumSource.save_not_exist(node_index,multi_keys);
            log.info("初始化订单生成数据库成功");
            return;
        }

        //有节点运行
        max_val.setId(null);
        max_val.setBase(max_val.getBase() + max_val.getUp_bound());
        max_val.setUp_bound(up_bound);
        var incr_key = UUID.randomUUID().toString();
        max_val.setIncr_key("xxx1:"+incr_key);
        max_val.setDate(new Date());
        max_val.setNode_id(Long.toString(node_index));
        max_val.setStatus(0l);

        //数据库的一条记录和redis中的记录 必须一致的

        //协作式并发
        //非协作式并发
        //状态一致性
        //消息安全性

        //状态一致性问题,影响时序,进而影响到恢复流程.所以,必须假定状态一致
        //消息安全也无法确定,所以消息也需要是假定安全

        //如果可以判断存在一致性问题后,我们就可以直接终止恢复.
        //状态不一致指的是:数据丢失,而非修改.例如,数据被部分删除了.
        //这种不一致应该只出现在redis,我们假定数据库总是一致的
        //不过,没有找到判断状态是否一致的方式

        var save_ok=false;
        try {
            save_ok=numSplitService.save(max_val);
        }catch (DuplicateKeyException e){
            //重复的原因可能是node_id与status列 唯一约束
            //或者 base列的唯一约束
            log.info("插入失败,检测并发插入,跳过插入继续");
            //处理恢复中断或卡住问题
            save_ok=true;
        }
        if (!save_ok) {
            log.error("数据库插入范围失败,稍后再试");
            //这里不是并发错误,是真正的错误,需要退出
            return;
        }

        //获取可能冲突的数据
        var one = numSplitService.getOne(new QueryWrapper<NumSplit>()
                .eq("status", 0)
                .eq("node_id",max_val.getNode_id())
                .eq("base",max_val.getBase())
        );
        //原则上,数据库不应该不一致.
        if(one!=null){//期待的情况
//            var multi_keys = Map.of(one.getIncr_key(), one.getUp_bound(), "sg_oss.incrby", one);
//            orderNumSource.save_not_exist(node_index, multi_keys);
            orderNumSource.save_not_exist(node_index, one.getIncr_key(), one.getUp_bound());
            if(orderNumSource.save_not_exist(node_index, "sg_oss.incrby", one)){
                log.info("null_incr恢复成功");
            }else{
                log.info("过期的恢复.其他进程恢复完成,当前进程退出");
            }
        }else{//在唯一索引上冲突了,下面要决定到底在哪个索引冲突了
            var onex = numSplitService.getOne(new QueryWrapper<NumSplit>()
                    .eq("status", 0)
                    .eq("node_id",max_val.getNode_id()));
            if(onex!=null){//node_id和status上冲突了
                orderNumSource.save_not_exist(node_index, onex.getIncr_key(), onex.getUp_bound());
                orderNumSource.save_not_exist(node_index, "sg_oss.incrby", onex);
                log.info("仅仅尝试恢复完成.此恢复目的是针对并发恢复过程中,之前恢复突然宕机的情况.");
            }else{//在base上冲突了
                log.info("当前节点的恢复未竞争到base,等待下次恢复");
            }
        }

    }
}
